package org.chnxi.study.messagequeue.kafka.comsumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class SendedToMsgDetailConsumer implements InitializingBean {

    @Autowired
    private Properties consumerConfig;

    @Value("${}")
    private String topicName;

    @Autowired
    private PollMsgProcessor<String, String> pollMsgProcessor;

    @Autowired
    private OffsetProcessor<String, String> offsetProcessor;

    @Override
    public void afterPropertiesSet() throws Exception {
        consumer();
    }

    public void consumer(){
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(consumerConfig);
        kafkaConsumer.subscribe(Collections.singletonList(topicName));
        kafkaConsumer.poll(500L);

        if(null != offsetProcessor){
            offsetProcessor.setKafkaConsumer(kafkaConsumer);
            offsetProcessor.resetOffset();
        }

        Thread mainThread = Thread.currentThread();

        // 启动处理拉取消息的线程
        Thread pollMsgThread = new Thread(() -> {
            try {
                pollMsgProcessor.pollMsg(kafkaConsumer);
            } finally {
                kafkaConsumer.close(3, TimeUnit.SECONDS);
            }
        }, "KAFKA_SENDEDTOMSGDETAIL");
        pollMsgThread.start();

        // 添加安全退出的监听
        Thread shutdownThread = new Thread(()->{
            kafkaConsumer.wakeup();
            try{
                mainThread.join();
            }catch (InterruptedException e){
                // TODO
            }
        }, "KAFKA_SENDEDTOMSGDETAIL_SHUTDOWNHOOK");
        Runtime.getRuntime().addShutdownHook(shutdownThread);
    }

    public Properties kafkaConsumerProps(){
        Properties props = new Properties();
        // 设置kafka服务地址
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "");
        // 设置session超时时长
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000);
        // 心跳检测间隔，受限于消息处理时长，要大于消息处理时长
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 60000);
        // 设置消费者组名称
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerGroup");
        // 设置key和value的反序列化处理器
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 关闭自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 设置初始化offset的方式
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 设置每次最大拉取消息数量
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        return props;
    }
}
